package concurrency.chapter9;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

/**
 * @author honway.liu
 * @date 2019-06-24 21:59
 * @email gm100861@gmail.com
 * @blog http://linuxsogood.org
 */
public class SimpleThreadPool{

    private final int poolSize;

    private static final int defaultSize = 10;

    private static final int maxQueueSize = 20;

    private static volatile boolean destroy = false;

    private List<WorkerThread> threads = new ArrayList<>();

    private final LinkedList<Runnable> tasks = new LinkedList<>();

    public static DiscardStrage defaultStrage = () -> {
        throw new DiscardException("discard.");
    };

    public static DiscardStrage discardStrage;

    public SimpleThreadPool() {
        this(defaultSize, defaultStrage);
    }

    public SimpleThreadPool(int poolSize, DiscardStrage strage) {
        this.poolSize = poolSize;
        discardStrage = strage;
        initThreadPool();
    }

    private void initThreadPool() {
        for (int i = 0; i < poolSize; i++) {
            WorkerThread workerThread = new WorkerThread(i + "");
            workerThread.start();
            threads.add(workerThread);
        }
    }

    /**
     * 提交任务到线程池
     * @param runnable
     */
    public void submit(Runnable runnable) {
        if (destroy) {
            throw new IllegalStateException("Thread is destroy, not allow submit task.");
        }
        // 为避免读写操作时的多线程安全问题,此处应用锁
        synchronized (tasks) {
            if (tasks.size() > maxQueueSize) {
                try {
                    discardStrage.discard();
                } catch (DiscardException e) {
                    // 队列超过了最大队列深度, 走拒绝策略
                    System.out.println(e.getMessage());
                }
            }
            tasks.addLast(runnable);
            tasks.notifyAll();
        }
    }

    /**
     * 关闭线程池
     * 只有当任务池中没有任务的时候,才去关闭线程池
     */
    public void shutdown() {
        while (!tasks.isEmpty()) {
            try {
                TimeUnit.MILLISECONDS.sleep(100);
                System.out.println("关闭线程池,但是还有任务,稍做等待再试,当前剩余任务数:" + tasks.size());
            } catch (InterruptedException e) {
                System.out.println("关闭线程池异常" + e.getMessage());
            }
        }
        // 任务队列为空,抢占锁,避免在关闭线程池的时候,任务还在提交
        synchronized (tasks) {
            int size = threads.size();
            System.out.println("关闭线程池, 当前线程池中线程数量:" + size);
            while (size > 0) {
                for (WorkerThread thread : threads) {
                    System.out.println("处理线程:" + thread.getThreadStatus());
                    if (thread.threadStatus == ThreadStatus.BLOCKED) {
                        thread.interrupt();
                        thread.close();
                        size--;
                    } else {
                        try {
                            TimeUnit.MILLISECONDS.sleep(100);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
            destroy = true;
            System.out.println("thread pool disposed");
        }
    }

    /**
     * 线程状态枚举
     */
    private enum ThreadStatus {
        FREE, RUNNING, BLOCKED, DEAD,
    }

    private class WorkerThread extends Thread {

        private volatile ThreadStatus threadStatus = ThreadStatus.FREE;

        private static final String THREAD_NAME_PREFIX = "SimpleThreadPool-";


        public WorkerThread(String name) {
            super(THREAD_NAME_PREFIX + name);
        }

        public ThreadStatus getThreadStatus() {
            return threadStatus;
        }

        @Override
        public void run() {
            while (threadStatus != ThreadStatus.DEAD) {
                Runnable task = null;
                synchronized (tasks) {
                    while (tasks.isEmpty()) {
                        try {
                            threadStatus = ThreadStatus.BLOCKED;
                            tasks.wait();
                        } catch (InterruptedException e) {
                            // 被中断,退出线程.
                            System.out.println(Thread.currentThread() + "检测到中断信号");
                            return;
                        }
                    }
                    task = tasks.removeFirst();
                }
                if (task != null) {
                    threadStatus = ThreadStatus.RUNNING;
                    task.run();
                    threadStatus = ThreadStatus.FREE;
                }

            }
        }

        public void close() {
            threadStatus = ThreadStatus.DEAD;
        }
    }


    private static class DiscardException extends Exception {

        public DiscardException(String message) {
            super(message);
        }
    }

    public interface DiscardStrage {
        void discard() throws DiscardException;
    }


    public static void main(String[] args) throws InterruptedException {
        SimpleThreadPool threadPool = new SimpleThreadPool();
        IntStream.rangeClosed(1, 30).forEach((i) -> {
            threadPool.submit(() -> {
                System.out.println(Thread.currentThread().getName() + " is start");
                try {
                    TimeUnit.SECONDS.sleep(2);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName() + " is finished.");
            });
        });

        TimeUnit.SECONDS.sleep(10);
        System.out.println("关闭线程池");
        threadPool.shutdown();
        threadPool.submit(() -> System.out.println("test shutdown thread pool, submit some task."));
    }
}
